package com.hzg.demo.broadcast;

import com.hzg.util.CommonUtil;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author: huangzhigao
 * @Date: 2022/4/20 20:57
 */
public class BroadCastDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = CommonUtil.getEnv();
        DataStreamSource<Action> source = env.fromElements(new Action());
        DataStreamSource<Pattern> rules = env.fromElements(new Pattern());
        KeyedStream<Action, Long> keyedStream = source.keyBy(Action::getUserId);
        MapStateDescriptor<Void, Pattern> mapState = new MapStateDescriptor<>("mapState", Types.VOID, Types.POJO(Pattern.class));
        BroadcastStream<Pattern> broadcastStream = rules.broadcast(mapState);
        SingleOutputStreamOperator<Tuple2<Long, Pattern>> process = keyedStream.connect(broadcastStream).process(new PatternEvaluator());
        process.print();
        env.execute();
    }

    public static class PatternEvaluator extends KeyedBroadcastProcessFunction<Long, Action, Pattern, Tuple2<Long, Pattern>> {

        //前一个事件的行为
        private ValueState<String> preActionState;
        //广播流事件
        private MapStateDescriptor<Void, Pattern> patternMapState;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            preActionState = getRuntimeContext().getState(new ValueStateDescriptor<String>("preActionState", String.class));
            patternMapState = new MapStateDescriptor("patternState", Types.VOID, Types.POJO(Pattern.class));

        }

        //处理真正的事件流
        @Override
        public void processElement(Action value, ReadOnlyContext ctx, Collector<Tuple2<Long, Pattern>> out) throws Exception {
            //从广播流中获取当前的广播状态
            Pattern pattern = ctx.getBroadcastState(patternMapState).get(null);
            String preAction = preActionState.value();
            if (pattern != null && preAction != null) {
                if (pattern.getFirstAction().equals(preAction) && pattern.getSecondAction().equals(value.getAction())) {
                    out.collect(Tuple2.of(ctx.getCurrentKey(), pattern));
                }
            }
            //更新当前的action
            preActionState.update(value.getAction());

        }

        //广播流，当再来一条广播数据，插入到状态中
        @Override
        public void processBroadcastElement(Pattern value, Context ctx, Collector<Tuple2<Long, Pattern>> out) throws Exception {
            BroadcastState<Void, Pattern> broadcastState = ctx.getBroadcastState(patternMapState);
            broadcastState.put(null, value);
        }
    }


}
